﻿Imports RabbitMQ.Client
Imports System.ComponentModel


Public Class RabbitEars
    Implements RabbitMQ.Client.IBasicConsumer


    Event Message(ByVal sender As Object, ByVal consumerTag As String, ByVal deliveryTag As ULong, ByVal redelivered As Boolean, ByVal exchange As String, ByVal routingKey As String, ByVal properties As RabbitMQ.Client.IBasicProperties, ByVal body() As Byte)

#Region "IBasicConsumer"
    Public Sub HandleBasicCancelOk(ByVal consumerTag As String) Implements RabbitMQ.Client.IBasicConsumer.HandleBasicCancelOk
        Console.WriteLine("Cancel ok")
    End Sub

    Public Sub HandleBasicConsumeOk(ByVal consumerTag As String) Implements RabbitMQ.Client.IBasicConsumer.HandleBasicConsumeOk
        Console.WriteLine("consume ok")
    End Sub

    Public Sub HandleModelShutdown(ByVal model As RabbitMQ.Client.IModel, ByVal reason As RabbitMQ.Client.ShutdownEventArgs) Implements RabbitMQ.Client.IBasicConsumer.HandleModelShutdown
        Console.WriteLine("shutdown")
        _model.Close()
    End Sub

    Public ReadOnly Property Model As RabbitMQ.Client.IModel Implements RabbitMQ.Client.IBasicConsumer.Model
        Get
            Return _model
        End Get
    End Property
#End Region

#Region "Event Helper"
    Shared Sub Raise(ByVal [event] As [Delegate], ByVal data As Object())
        If [event] IsNot Nothing Then
            For Each C In [event].GetInvocationList
                Dim T = CType(C.Target, ISynchronizeInvoke)
                If T IsNot Nothing AndAlso T.InvokeRequired Then T.BeginInvoke(C, data) Else C.DynamicInvoke(data)
            Next
        End If
    End Sub
#End Region

    Public Sub HandleBasicDeliver(ByVal consumerTag As String, ByVal deliveryTag As ULong, ByVal redelivered As Boolean, ByVal exchange As String, ByVal routingKey As String, ByVal properties As RabbitMQ.Client.IBasicProperties, ByVal body() As Byte) Implements RabbitMQ.Client.IBasicConsumer.HandleBasicDeliver

        Raise(MessageEvent, New Object() {Me, consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body})

        _model.BasicAck(deliveryTag, False)

        If Throttle > 0 Then Threading.Thread.Sleep(Throttle)
    End Sub

#Region "Properties"
    Public Property ExchangeName As String = ""
    Public Property ExchangeWhatType As String = ExchangeType.Direct
    Public Property QueueName As String = ""
    Public Property RoutingKey As String = ""
    Public Property HostName As String = ""
    Public Property VirtualHost As String = "/"
    Public Property Protocol As RabbitMQ.Client.IProtocol = Protocols.AMQP_0_9_1
    Public Property Port As Integer = AmqpTcpEndpoint.UseDefaultPort
    Public Property NoAck As Boolean = False
    Public Property Throttle As Integer = 0
    Public Property Durable As Boolean = False

    Private _conn As RabbitMQ.Client.IConnection
    Private _model As RabbitMQ.Client.IModel

#End Region

#Region "Constructors"
    Public Sub New(ByRef TheModel As IModel, ByVal TheQueueName As String)
        _model = TheModel
        QueueName = TheQueueName
    End Sub

    Public Sub New()

    End Sub

    Public Sub New(ByVal exchangeName As String, ByVal exchangeWhatType As String, ByVal queueName As String, ByVal routingKey As String, ByVal hostName As String, ByVal virtualHost As String, ByVal protocol As RabbitMQ.Client.IProtocol, ByVal port As Integer, ByVal noAck As Boolean, ByVal throttle As Integer, ByVal durable As Boolean, ByVal model As RabbitMQ.Client.IModel)
        Me.ExchangeName = exchangeName
        Me.ExchangeWhatType = exchangeWhatType
        Me.QueueName = queueName
        Me.RoutingKey = routingKey
        Me.HostName = hostName
        Me.VirtualHost = virtualHost
        Me.Protocol = protocol
        Me.Port = port
        Me.NoAck = noAck
        Me.Throttle = throttle
        Me.Durable = durable
        _model = model
    End Sub
#End Region

    Sub Open(Optional ByVal Consume As Boolean = True)

        If _model Is Nothing Then
            Dim factory As ConnectionFactory = New ConnectionFactory
            factory.VirtualHost = VirtualHost
            factory.Protocol = Protocol
            factory.HostName = HostName
            factory.Port = Port

            _conn = factory.CreateConnection

            _model = _conn.CreateModel
            _model.ExchangeDeclare(ExchangeName, ExchangeWhatType, Durable)
            _model.QueueDeclare(QueueName)
            _model.QueueBind(QueueName, ExchangeName, RoutingKey, NoAck, Nothing)
        End If

        If Consume = True Then _model.BasicConsume(QueueName, Nothing, Me)

    End Sub

    Sub Close()
        _conn.Close()
    End Sub


#Region "Send"
    Sub Send(ByVal Message As String, Optional ByVal UseRoutingKey As String = "")
        Dim MessagebOdyBytes As Byte() = System.Text.Encoding.UTF8.GetBytes(Message)
        _model.BasicPublish(ExchangeName, UseRoutingKey, Nothing, MessagebOdyBytes)
    End Sub

    Sub Send(ByVal MessageBytes As Byte(), Optional ByVal UseRoutingKey As String = "")
        _model.BasicPublish(ExchangeName, UseRoutingKey, Nothing, MessageBytes)
    End Sub
#End Region


End Class